package com.changan.common.parser;

import com.changan.common.model.Tuple2;

import java.util.List;
import java.util.Properties;

/**
 * @Author: wmh
 * Create Time: 2021/6/3 17:19
 */
public class FlinkSqlParse {
    public static String sqlParse(String tableName, List<Tuple2<String, String>> sqlField, Properties properties, List<Tuple2<String, String>> partition, String primaryKeysNotEnforced) {
        return "CREATE TABLE " + tableName + " (\n" +
                sqlField.stream()
                        .map((x) -> x._1.startsWith("PRIMARY") ? "    " + x._1 + " " + x._2.toUpperCase() : "    `" + x._1 + "` " + x._2.toUpperCase())
                        .reduce(
                                (x, y) -> x + ",\n" + y
                        ).orElse("") + (primaryKeysNotEnforced == null ? "" : ",\n     PRIMARY KEY(" + primaryKeysNotEnforced + ") NOT ENFORCED ") +
                (partition != null ? ",\n" + partition.stream().map((x) -> x._1 + " " + x._2).reduce((x, y) -> x + ",\n" + y).orElse("") : "") + "\n)" +
                (partition != null ? "PARTITIONED BY (" + partition.stream().map(x -> x._1).reduce((x, y) -> "`" + x + "`" + ",`" + y + "`").get() + ")" : "") + " WITH (\n" +
                properties.entrySet().stream().map(x -> "    '" + x.getKey() + "' = '" + x.getValue() + "'").reduce((x, y) -> x + ",\n" + y).orElse("") + "\n)";
    }

    public static String sqlParseWithNoPart(String tableName, List<Tuple2<String, String>> sqlField, Properties properties) {
        return sqlParse(tableName, sqlField, properties, null, null);
    }

    public static String sqlParseWithNoPart(String tableName, List<Tuple2<String, String>> sqlField, Properties properties, String primaryKeysNotEnforced) {
        return sqlParse(tableName, sqlField, properties, null, primaryKeysNotEnforced);
    }
}
